package biz

import (
	"context"
	consul "github.com/go-kratos/consul/registry"
	"github.com/go-kratos/kratos/v2/log"
	"github.com/go-kratos/kratos/v2/transport/grpc"
	"github.com/hashicorp/consul/api"
	"github.com/zhenjl/cityhash"
	"go_private_im/api/logic"
	"go_private_im/api/protocol"
	"go_private_im/internal/comet/conf"
	"go_private_im/internal/comet/pkg"
	"time"
)

type CometUsecase struct {
	c         *conf.Config
	log       *log.Helper
	buckets   []*pkg.Bucket // subkey bucket
	bucketIdx uint32
	rpcClient logic.LogicClient
}

// Buckets return all buckets.
func (c *CometUsecase) Buckets() []*pkg.Bucket {
	return c.buckets
}

func newLogicClient(c *conf.Consul) logic.LogicClient {
	config := api.DefaultConfig()
	config.Address = c.Addr
	client, err := api.NewClient(config)
	if err != nil {
		panic(err)
	}
	dis := consul.New(client)
	endpoint := "discovery://default/private.logic"
	conn, err := grpc.DialInsecure(context.Background(), grpc.WithEndpoint(endpoint), grpc.WithDiscovery(dis))
	if err != nil {
		panic(err)
	}

	return logic.NewLogicClient(conn)
}


func NewCometUsecase(config *conf.Config,logger log.Logger) *CometUsecase {
	comets := &CometUsecase{
		c:         config,
		log:       log.NewHelper(logger),
		rpcClient: newLogicClient(config.Consul),
	}
	comets.buckets = make([]*pkg.Bucket, config.Bucket.Size)

	comets.bucketIdx = uint32(config.Bucket.Size)

	for i := 0; i < config.Bucket.Size; i++ {
		comets.buckets[i] = pkg.NewBucket(config.Bucket)
	}
	return comets
}

type AuthInfo struct {
	Ch *pkg.Channel
	Bucket *pkg.Bucket
	UserId int64
	Key string
	Accepts []int32
	Hb time.Duration
	RoomId string
}

func (c *CometUsecase) Auth (ctx context.Context,auth *protocol.Proto) (*AuthInfo,error) {
	req := logic.ConnectReq{Server: c.c.Env.Host, Token: auth.Body}

	connect, err := c.rpcClient.Connect(ctx, &req)
	if err != nil {
		c.log.Error(err)
		return nil, err
	}
	channel := pkg.NewChannel(c.c.Protocol.CliProto, c.c.Protocol.SvrProto)
	channel.Watch(connect.Accepts...)
	channel.Key = connect.Key
	channel.UserId = connect.UserId
	bucket := &pkg.Bucket{}
	bucket = c.Bucket(channel.Key)
	err = bucket.Put(connect.RoomId, channel)
	if err != nil {
		c.log.Infof("websocket connnected key:%s mid:%d proto:%+v", channel.Key, channel.UserId, auth)
		return nil, err
	}
	authInfo := &AuthInfo{
		Ch:      channel,
		Bucket:  bucket,
		UserId:     connect.UserId,
		Hb:      time.Duration(connect.Heartbeat),
		Accepts: connect.Accepts,
		Key:     connect.Key,
		RoomId: connect.RoomId,
	}
	return authInfo, nil
}

func (c *CometUsecase) Receive (ctx context.Context,authInfo *AuthInfo,proto *protocol.Proto)  (err error) {
	_, err = c.rpcClient.Receive(ctx, &logic.ReceiveReq{UserId: authInfo.UserId, Proto: proto})
	return
}
// Bucket get the bucket by subkey.
func (c *CometUsecase) Bucket(subKey string) *pkg.Bucket {
	idx := cityhash.CityHash32([]byte(subKey), uint32(len(subKey))) % c.bucketIdx
	return c.buckets[idx]
}


//上报
func (c *CometUsecase) onlineproc() {
	for {
		var (
			allRoomsCount map[string]int32
			err           error
		)
		roomCount := make(map[string]int32)
		for _, bucket := range c.buckets {
			for roomId, count := range bucket.RoomsCount() {
				roomCount[roomId] += count
			}
		}
		if allRoomsCount, err = c.RenewOnline(context.Background(), roomCount); err != nil {
			time.Sleep(time.Second)
			continue
		}
		for _, bucket := range c.buckets {
			bucket.UpRoomsCount(allRoomsCount)
		}
		time.Sleep(time.Second * 10)
	}

}
// RenewOnline renew room online.
func (c *CometUsecase) RenewOnline(ctx context.Context, roomCount map[string]int32) (allRoom map[string]int32, err error) {
	reply, err := c.rpcClient.RenewOnline(ctx, &logic.OnlineReq{
		Server:    c.c.Env.Host,
		RoomCount: roomCount,
	})
	if err != nil {
		return
	}
	return reply.AllRoomCount, nil
}

func (c *CometUsecase) Heartbeat(ctx context.Context,authInfo *AuthInfo) (err error)  {
	_, err = c.rpcClient.Heartbeat(ctx, &logic.HeartbeatReq{
		Server: c.c.Env.Host,
		UserId:    authInfo.UserId,
		Key:    authInfo.Key,
	})
	return
}
